package com.dingwen.treasure.kettle.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.dingwen.treasure.kettle.constant.TaskConstant;
import com.dingwen.treasure.kettle.dto.modify.TaskInfoDto;
import com.dingwen.treasure.kettle.dto.modify.TaskInfoStateDto;
import com.dingwen.treasure.kettle.dto.query.TaskInfoQueryDto;
import com.dingwen.treasure.kettle.entity.TaskInfo;
import com.dingwen.treasure.kettle.enums.FileTypeEnum;
import com.dingwen.treasure.kettle.enums.TaskStateEnum;
import com.dingwen.treasure.kettle.enums.TaskTypeEnum;
import com.dingwen.treasure.kettle.service.TaskInfoService;
import com.dingwen.treasure.kettle.task.manager.TaskManager;
import com.dingwen.treasure.kettle.utils.EnumUtil;
import com.dingwen.treasure.kettle.vo.FileVo;
import com.dingwen.treasure.kettle.vo.TaskInfoVo;
import com.dingwen.treasure.kettle.execute.KettleExecute;
import com.dingwen.treasure.kettle.mapper.TaskInfoMapper;
import com.dingwen.treasure.kettle.service.FileService;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.repository.RepositoryObjectType;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * TaskInfoServiceImpl
 *
 * @author dingwen
 * @date 2022/08/22
 */
@Service
@Slf4j
public class TaskInfoServiceImpl extends ServiceImpl<TaskInfoMapper, TaskInfo> implements TaskInfoService {


    @Resource
    private TaskInfoMapper taskInfoMapper;


    @Resource
    private FileService fileService;

    /**
     * kettle执行
     */
    @Resource
    private KettleExecute kettleExecute;

    /**
     * 任务管理器
     */
    @Lazy
    @Resource
    private TaskManager taskManager;

    /**
     * 成功次数 +1
     *
     * @param taskId 任务id
     */
    @Override
    public void incrementForSuccess(Long taskId) {
        TaskInfo taskInfo = getById(taskId);
        taskInfo.setSuccess(taskInfo.getSuccess() + 1);
        updateById(taskInfo);
    }

    /**
     * 失败次数 +1
     *
     * @param taskId 任务id
     */
    @Override
    public void incrementForFail(Long taskId) {
        TaskInfo taskInfo = getById(taskId);
        taskInfo.setFail(taskInfo.getFail() + 1);
        updateById(taskInfo);
    }

    /**
     * 刷新
     */
    @Override
    @Async("asyncExecutor")
    public void refresh() {
        // 获取所有不存在任务的包括转换、作业，当前不存在的任务信息进行一次性任务初始化

        // 资源库存在的文件Id
        Map<String, List<FileVo>> fileMaps = fileService.getAllFile()
                .parallelStream()
                .collect(Collectors.groupingBy(FileVo::getType));

        if (!fileMaps.containsKey(RepositoryObjectType.TRANSFORMATION.getTypeDescription())
                || !fileMaps.containsKey(RepositoryObjectType.JOB.getTypeDescription())) {
            log.error("刷新失败，资源库文件信息获取异常");
            return;
        }
        List<String> fileIdsForTrans = fileMaps.get(RepositoryObjectType.TRANSFORMATION.getTypeDescription())
                .parallelStream()
                .map(FileVo::getFileId)
                .collect(Collectors.toList());
        List<String> fileIdsForJob = fileMaps.get(RepositoryObjectType.JOB.getTypeDescription())
                .parallelStream()
                .map(FileVo::getFileId)
                .collect(Collectors.toList());

        LambdaQueryWrapper<TaskInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.select(TaskInfo::getFileId, TaskInfo::getFileType);

        // 任务表存在的文件Id
        Map<FileTypeEnum, List<TaskInfo>> taskFileIds = list(queryWrapper)
                .parallelStream()
                .collect(Collectors.groupingBy(TaskInfo::getFileType));
        if (CollUtil.isNotEmpty(taskFileIds)) {
            List<String> existJobIds = taskFileIds.get(FileTypeEnum.JOB)
                    .parallelStream()
                    .map(TaskInfo::getFileId)
                    .collect(Collectors.toList());

            List<String> existTransIds = taskFileIds.get(FileTypeEnum.TRANS)
                    .parallelStream()
                    .map(TaskInfo::getFileId)
                    .collect(Collectors.toList());
            // 求差集
            if (CollUtil.isNotEmpty(existJobIds)) {
                fileIdsForJob.removeAll(existJobIds);
            }
            if (CollUtil.isNotEmpty(existTransIds)) {
                fileIdsForTrans.removeAll(existTransIds);
            }
        }


        saveBatch(buildTaskInfo(fileIdsForJob, RepositoryObjectType.JOB));
        saveBatch(buildTaskInfo(fileIdsForTrans, RepositoryObjectType.TRANSFORMATION));

        log.info("刷新成功，共计生成任务：{}条", fileIdsForJob.size() + fileIdsForTrans.size());

    }

    /**
     * 分页查询任务列表
     *
     * @param taskInfoQueryDto 任务信息dto
     * @return {@link IPage}
     */
    @Override
    public IPage<TaskInfoVo> getTaskInfoPage(TaskInfoQueryDto taskInfoQueryDto) {
        Integer current = Optional.ofNullable(taskInfoQueryDto.getCurrent()).orElse(1);
        Integer pageSize = Optional.ofNullable(taskInfoQueryDto.getPageSize()).orElse(10);
        return taskInfoMapper.getTaskInfoPage(new Page<>(current, pageSize), taskInfoQueryDto);
    }

    /**
     * 立即执行
     *
     * @param taskId 任务id
     */
    @Override
    public void executeImmediately(Long taskId) {
        TaskInfoVo taskInfoVo = getByTaskId(taskId);
        switch (taskInfoVo.getTaskType()) {
            case ONE_TIME_TASK:
                kettleExecute.exe(taskInfoVo.getFileType(), taskInfoVo.getFileFullName());
                break;
            case PERIODIC_TASK:
                taskManager.startJobNow(taskId);
                break;
            default:
                break;

        }

    }

    /**
     * 添加周期任务
     *
     * @param taskInfoDto 任务信息dto
     * @return {@link TaskInfo}
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public TaskInfoVo addPeriodicTask(TaskInfoDto taskInfoDto) {
        TaskInfo taskInfo = taskInfoDto.dtoToPO(taskInfoDto);
        return taskInfoDto.poToVO(taskManager.addJob(taskInfo, true));
    }

    /**
     * 获取所有正在执行中的周期任务
     *
     * @return {@link List}<{@link TaskInfoVo}>
     */
    @Override
    public List<Map<String, Object>> getRunningPeriodicTaskAll() {
        return taskManager.getAllJob();
    }

    /**
     * 通过任务Id修改任务信息
     *
     * @param taskInfoDto 任务信息dto
     */
    @Override
    public void modifyTaskInfoByTaskId(TaskInfoDto taskInfoDto) {
        TaskInfo taskInfo = taskInfoDto.dtoToPO(taskInfoDto);
        Assert.notNull(taskInfo, "任务不存在");
        switch (taskInfoDto.getTaskType()) {
            // 一次性任务
            case ONE_TIME_TASK:
                updateById(taskInfo);
                break;
            case PERIODIC_TASK:
                // 周期性任务
                TaskInfo adapterTaskInfo = taskManager.adapterTaskInfo(taskInfo);
                taskManager.modifyJob(adapterTaskInfo);
                break;
            default:
                break;
        }
    }

    /**
     * 通过任务Id删除任务信息
     *
     * @param taskId   任务id
     * @param taskType 任务类型：1一次性任务 2周期性任务
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public void removeTaskInfoByTaskId(Long taskId, Integer taskType) {
        switch (EnumUtil.getEnumByCode(TaskTypeEnum.class, taskType)) {
            // 一次性任务
            case ONE_TIME_TASK:
                removeById(taskId);
                break;
            case PERIODIC_TASK:
                // 周期性任务
                TaskInfo taskInfo = getById(taskId);
                taskManager.deleteJob(taskInfo);
                removeById(taskInfo.getTaskId());
                break;
            default:
                break;
        }
    }

    /**
     * 修改任务状态
     *
     * @param taskInfoStateDto 任务信息状态dto
     */
    @Override
    public void modifyTaskState(TaskInfoStateDto taskInfoStateDto) {
        taskManager.modifyJobState(taskInfoStateDto);
    }

    /**
     * 通过任务id查询TaskInfoVo
     *
     * @param taskId 任务id
     * @return {@link TaskInfoVo}
     */
    @Override
    public TaskInfoVo getByTaskId(Long taskId) {
        TaskInfoQueryDto queryDto = TaskInfoQueryDto.builder()
                .taskId(taskId)
                .pageSize(1)
                .current(1)
                .build();
        IPage<TaskInfoVo> taskInfoPage = getTaskInfoPage(queryDto);
        Assert.isTrue(Objects.nonNull(taskInfoPage) && taskInfoPage.getRecords().size() == 1, "任务信息获取异常");
        return taskInfoPage.getRecords().get(0);
    }

    /**
     * 构建需要系统生成的任务信息
     *
     * @param fileIds 文件id
     * @return {@link List}<{@link TaskInfo}>
     */
    private List<TaskInfo> buildTaskInfo(List<String> fileIds, RepositoryObjectType repositoryObjectType) {
        List<TaskInfo> tasks = new ArrayList<>(fileIds.size());
        fileIds.parallelStream()
                .forEach(fileId -> {
                    LocalDateTime now = LocalDateTime.now();
                    TaskInfo taskInfo = TaskInfo.builder()
                            .taskType(TaskTypeEnum.ONE_TIME_TASK)
                            .fileId(fileId)
                            .taskCode(TaskConstant.TASK_CODE_SYSTEM_GENERATE)
                            .jobName(TaskConstant.TASK_NAME_SYSTEM_GENERATE)
                            .taskState(TaskStateEnum.ENABLE)
                            .createTime(now)
                            .updateTime(now)
                            .fileType(
                                    repositoryObjectType.getTypeDescription().equals(RepositoryObjectType.TRANSFORMATION.getTypeDescription())
                                            ? FileTypeEnum.TRANS : FileTypeEnum.JOB)
                            .build();
                    tasks.add(taskInfo);
                });
        return tasks;
    }
}




